package org.joy.db2.query;

import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.Closeable;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.Semaphore;

import org.joy.analyzer.TokenAnalyzer;
import org.joy.db2.ResultEntry;
import org.joy.nlp.Word;
import org.joy.nlp.WordSpliter;

public class SocketClient {
    private ArrayList<DBServer> servers = new ArrayList<DBServer>();
    private static TokenAnalyzer tokenizer = new TokenAnalyzer();

    public static void setTokenizer(TokenAnalyzer tokenizer) {
	SocketClient.tokenizer = tokenizer;
    }

    public static TokenAnalyzer getTokenizer() {
	return tokenizer;
    }

    private HashMap<String, List<ResultEntry>> cache = new HashMap<String, List<ResultEntry>>();

    public SocketClient(String[] severs) throws SocketException {
	for (String s : severs) {
	    servers.add(new DBServer(new InetSocketAddress(s, 1987)));
	}
    }

    static class DBServer implements Closeable {
	private DatagramSocket s;
	private InetSocketAddress serverAddress;
	/**
	 * PAGE SIZE MUST SMALLER THAN BLOCK SIZE!!!
	 */
	private final static int BLOCK_SIZE = 10;
	private HashMap<String, Queue<ResultEntry>> cache = new HashMap<String, Queue<ResultEntry>>();

	public DBServer(InetSocketAddress serverAddress) throws SocketException {
	    s = new DatagramSocket();
	    this.serverAddress = serverAddress;
	}

	public void load(String segmentedQuery) throws IOException {
	    // connect to db server
	    s.connect(serverAddress);
	    // count how many to load
	    int count = BLOCK_SIZE
		    - (cache.get(segmentedQuery) == null ? 0 : cache.get(
			    segmentedQuery).size());
	    if (count == 0)
		return;
	    // send the query
	    String line = "s " + count + " " + segmentedQuery;
	    s.send(new DatagramPacket(line.getBytes("utf-8"), line.getBytes("utf-8").length));

	    // read the reply
	    DatagramPacket p = new DatagramPacket(new byte[4096], 4096);
	    s.receive(p);
	    byte[] data = new byte[p.getLength()];
	    System.arraycopy(p.getData(), p.getOffset(), data, 0, data.length);
	    // parse the reply
	    BufferedReader br = new BufferedReader(new InputStreamReader(
		    new ByteArrayInputStream(data),"utf-8"));
	    Queue<ResultEntry> resQueue = new LinkedList<ResultEntry>();
	    line = br.readLine();
	    while (line != null) {
		resQueue.offer(new ResultEntry(line));
		line = br.readLine();
	    }
	    cache.put(segmentedQuery, resQueue);
	}

	public ResultEntry next(String s) throws IOException {
	    // check queue
	    // if (cache.get(s) == null || cache.get(s).size() == 0) {
	    // load(s);
	    // }
	    return cache.get(s).poll();
	}

	public Double getTopScore(String s) throws IOException {
	    // check queue
	    // if (cache.get(s) == null || cache.get(s).size() == 0) {
	    // load(s);
	    // }
	    if (cache.get(s).size() == 0)
		return Double.MIN_VALUE;
	    return cache.get(s).peek().getScore();
	}

	@Override
	public void close() throws IOException {
	    s.close();
	}
    }

    public List<ResultEntry> search(final String query, final int count)
	    throws InterruptedException, IOException {
	final ArrayList<ResultEntry> ret = new ArrayList<ResultEntry>();
	final Semaphore s = new Semaphore(-servers.size() + 1);
	// token
	tokenizer.setDoc(new Query(query));
	tokenizer.doAnalyze();
	StringBuffer sb = new StringBuffer();
	for (Word w : tokenizer.output()) {
	    sb.append(w.getText() + " ");
	}
	final String segmentedQuery = sb.toString().trim();
	for (final DBServer server : servers) {
	    Thread t = new Thread(new Runnable() {
		@Override
		public void run() {
		    try {
			server.load(segmentedQuery);
			s.release();
		    } catch (Exception e) {
			e.printStackTrace();
		    }
		}
	    });
	    t.start();
	}
	s.acquire();

	while (ret.size() < count) {
	    double minScore = Double.MAX_VALUE;
	    int index = 0;
	    // find the next min score
	    for (int i = 0; i < servers.size(); i++) {
		if (minScore > servers.get(i).getTopScore(segmentedQuery)) {
		    minScore = servers.get(i).getTopScore(segmentedQuery);
		    index = i;
		}
	    }
	    if (minScore == Double.MIN_VALUE)
		break;
	    ret.add(servers.get(index).next(segmentedQuery));
	}
	// append them to the end of last query result
	cache.get(query).addAll(ret);
	return ret;
    }

    public List<ResultEntry> search(String query, int page, int pageSize)
	    throws InterruptedException, IOException {
	int start = page * pageSize, end = start + pageSize;
	if (cache.get(query) == null)
	    cache.put(query, new ArrayList<ResultEntry>());
	// check the cache
	if (end > cache.get(query).size()) {
	    search(query, end - cache.get(query).size());
	}
	return cache.get(query).subList(start,
		end > cache.get(query).size() ? cache.get(query).size() : end);
    }

    public static void main(String[] args) throws IOException,
	    InstantiationException, IllegalAccessException,
	    ClassNotFoundException, InterruptedException {
	String spliterName = args.length >= 2 ? args[1]
		: "org.joy.nlp.LuceneWordSpliter";
	SocketClient client = new SocketClient(args[0].split(":"));
	SocketClient.tokenizer.input((WordSpliter) Class.forName(spliterName)
		.newInstance());

	BufferedReader reader = new BufferedReader(new InputStreamReader(
		System.in));
	String line = reader.readLine();
	while (line != null) {
	    List<ResultEntry> res = client.search(line.split(";")[0], Integer
		    .parseInt(line.split(";")[1]), 5);
	    for (ResultEntry r : res) {
		System.out.println(r);
	    }
	    line = reader.readLine();
	}
	reader.close();
    }
}
